package com.gator.rabbitmq.pablo_rabbitmq.publish_subscription;


import com.gator.rabbitmq.pablo_rabbitmq.utils.ConnectionUtils;
import com.rabbitmq.client.CancelCallback;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;

/**
 * @Author PABLO
 * @Date 2022/6/4 22:11
 * @Desc 发布订阅模式，一个数据可通过交换机进行分发
 */
public class Publish_Subscription_Consumer {

    public static final String EXCHANGE_NAME = "pablo-exchange-fanout";
    public static final String EXCHANGE_TYPE = "fanout";

    public static void main(String[] args) {
        receiveMsg();
    }


    /**
     * @Description:消费者通过交换机消费数据
     * @Author: PABLO
     * @Date: 2022/6/5 11:42
     * @Params: []
     * @Return: void
     **/
    public static void receiveMsg() {
        Channel channel = ConnectionUtils.getConnectionAndChannel();
        try {
            //声明交换机,类型为扇出广播
            channel.exchangeDeclare(EXCHANGE_NAME, EXCHANGE_TYPE);
            //声明队列,这里声明随机队列,消费者断开连接后该队列就删除了
            String queue = channel.queueDeclare().getQueue();
            //交换机绑定队列
            channel.queueBind(queue, EXCHANGE_NAME, "");

            try {
                //消费
                DeliverCallback deliverCallback = (consumerTag, message) -> {
                    System.out.println(new String(message.getBody()));
                };
                //消费者取消消费
                CancelCallback cancelCallback = consumerTag -> {
                    System.out.println("msg被中断...");
                };
                //消费 自动确认
                channel.basicConsume(queue, true, deliverCallback, cancelCallback);
            } catch (Exception e) {
                e.printStackTrace();
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

class Publish_Subscription_Consumer2 {
    public static final String EXCHANGE_NAME = "pablo-exchange-fanout";
    public static final String EXCHANGE_TYPE = "fanout";

    public static void main(String[] args) {
        receiveMsg();
    }


    /**
     * @Description:消费者通过交换机消费数据
     * @Author: PABLO
     * @Date: 2022/6/5 11:42
     * @Params: []
     * @Return: void
     **/
    public static void receiveMsg() {
        Channel channel = ConnectionUtils.getConnectionAndChannel();
        try {
            //声明交换机,类型为扇出广播
            channel.exchangeDeclare(EXCHANGE_NAME, EXCHANGE_TYPE);
            //声明队列,这里声明随机队列,消费者断开连接后该队列就删除了
            String queue = channel.queueDeclare().getQueue();
            //交换机绑定队列
            channel.queueBind(queue, EXCHANGE_NAME, "");

            try {
                //消费
                DeliverCallback deliverCallback = (consumerTag, message) -> {
                    System.out.println(new String(message.getBody()));
                };
                //消费者取消消费
                CancelCallback cancelCallback = consumerTag -> {
                    System.out.println("msg被中断...");
                };
                //消费 自动确认
                channel.basicConsume(queue, true, deliverCallback, cancelCallback);
            } catch (Exception e) {
                e.printStackTrace();
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
